package com.diuse.config;

import com.diuse.config.communicate.HeartBeatDo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.Socket;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * 单例模式,这里的长连接只发送心跳包
 * Created by carey on 2016/11/19 0019.
 */
public class ClientManager {


    private static ClientManager instance;

    private Server server = Server.getInstance();
    private volatile boolean isStop = false;
    private final long receiveTimeDelay = 3000;
    private final long sleepTime = receiveTimeDelay;
    private ConcurrentHashMap<String, LinkedHashSet<HeartBeatHolder>> clientMap = new ConcurrentHashMap<String, LinkedHashSet<HeartBeatHolder>>();

    private static final Logger logger = LoggerFactory.getLogger(ClientManager.class);

    private ClientManager() {

    }


    /**
     * 启动管理线程
     */
    private void start() {
        new Thread(new CleanWorker()).start();
    }


    /**
     * 有新的client链接进入
     *
     * @param heartBeatDo
     * @param clientSocket
     */
    public void addClient(HeartBeatDo heartBeatDo, Socket clientSocket) {
        if (null == clientSocket || null == heartBeatDo) {
            return;
        }

        boolean doAdd;
        HeartBeatHolder heartBeatHolder = new HeartBeatHolder(clientSocket);
        String appName = heartBeatDo.getAppName();
        logger.info("addclient, " + appName);
        LinkedHashSet<HeartBeatHolder> clients = clientMap.get(appName);

        if (null == clients) {
            clients = new LinkedHashSet<HeartBeatHolder>();
            clients.add(heartBeatHolder);
            doAdd = true;
            LinkedHashSet<HeartBeatHolder> result = clientMap.putIfAbsent(appName, clients);
            //之前并发写入了
            if (null != result) {
                synchronized (result) {
                    doAdd = result.add(heartBeatHolder);
                }
            }
        } else {
            synchronized (clients) {
                doAdd = clients.add(heartBeatHolder);
            }
        }

        if (doAdd) {
            notifyClient(appName);
        }

    }

    /**
     *
     */
    private void removeTime() {
        while (true) {
            System.out.println("start to check overtime");
            for (Map.Entry<String, LinkedHashSet<HeartBeatHolder>> appClients : clientMap.entrySet()) {
                String appName = appClients.getKey();
                boolean needNotity = false;
                Iterator<HeartBeatHolder> itor = appClients.getValue().iterator();
                while (itor.hasNext()) {
                    HeartBeatHolder heartBeatHolder = itor.next();
                    Socket heartBeatSocket = heartBeatHolder.getSocket();
                    try {
                        InputStream in = heartBeatSocket.getInputStream();
                        if (in.available() > 0) {
                            ObjectInputStream ois = new ObjectInputStream(in);
                            Object obj = ois.readObject();
                            heartBeatHolder.setLastCheckTime(System.currentTimeMillis());
                            System.out.println("接收：\t" + obj);
                            ObjectOutputStream oos = new ObjectOutputStream(heartBeatSocket.getOutputStream());
                            oos.writeObject(obj);
                            oos.flush();
                        } else {
                            if (System.currentTimeMillis() - heartBeatHolder.getLastCheckTime() > receiveTimeDelay) {
                                itor.remove();
                                needNotity = true;
                            }
                        }
                    } catch (Exception e) {
                        logger.error("io e", e);
                        e.printStackTrace();
                        itor.remove();
                        needNotity = true;
                    }

                }

                if (needNotity) {
                    notifyClient(appName);
                }
            }
            try{
                Thread.sleep(sleepTime);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 通知集群的机子有数量有变化
     * 如果某个机子当前推送有问题，现在不关心，等到下一次心跳的时候自动失效
     *
     * @param appName 应用名称
     */
    private void notifyClient(String appName) {
        logger.info("start to notify: " + appName);
        System.out.println("start to notify: " + appName);
        LinkedHashSet<HeartBeatHolder> clients = clientMap.get(appName);
        Set<HeartBeatHolder> safeClients = new CopyOnWriteArraySet<HeartBeatHolder>(clients);
        int i =1;
        for (HeartBeatHolder clientHolder : safeClients) {
            Socket socket = clientHolder.getSocket();
            ObjectOutputStream oos = null;
            try {
                oos = new ObjectOutputStream(socket.getOutputStream());
                oos.writeObject(String.format("total:%s;index:%s",safeClients.size(),i++));
                oos.flush();
            } catch (IOException e) {
                logger.error("ioe",e );
                e.printStackTrace();
            }

        }
    }

    /**
     * 获取单例
     *
     * @return
     */
    public static ClientManager getInstance() {
        if (null != instance) {
            return instance;
        }
        synchronized (ClientManager.class) {
            if (null != instance) {
                return instance;
            }
            instance = new ClientManager();

            instance.start();
            logger.info("start timeout thread");
            return instance;
        }
    }


    /**
     * 清除已经超时的连接
     */
    @SuppressWarnings("InfiniteLoopStatement")
    class CleanWorker implements Runnable {

        //这里不关心server是否已经停了，只关心内部管理的socket

        public void run() {
            while (true) {
                removeTime();
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    logger.error("睡眠异常", e);
                }
            }
        }
    }
}



